<?php
/**
 * Created by PhpStorm.
 * User: huangj11
 * Date: 2020/12/30
 * Time: 19:09
 */

namespace console\controllers\bigdata;

use business\base\caches\DataCache;
use business\bigdata\services\queues\SupplierConsumerService;
use business\bigdata\services\queues\SupplierProducerService;
use common\queue\QueueConstants;
use iQueue\controllers\ShellConsumeController;

class SupplierController extends ShellConsumeController
{

    protected $queueAction = 'bigdata/supplier/process';

    public function actionFull()
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $service = new SupplierProducerService();
            $service->full();
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    public function actionIncrement()
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $service = new SupplierProducerService();
            $service->increment();
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }


    public function actionSingle($supplierId)
    {
        if (empty($supplierId)) {
            echo '供应商ID不能为空' . PHP_EOL;
        }
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $service = new SupplierProducerService();
            $service->single($supplierId);
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 消费端实现
     * php yii bigdata/supplier/consume 'bigdata-supplier-index-queue' true
     * docker exec -i opensearch_php php /var/www/html/opensearch/protected/yii bigdata/supplier/consume 'bigdata-supplier-index-queue' true
     * @param $message
     * @param $messageId
     */
    public function actionProcess($message, $messageId)
    {
        echo '正在执行:' . PHP_EOL;
        echo 'message: ' .$message. PHP_EOL;
        echo 'messageId: ' .$messageId. PHP_EOL;
        try {
            $cacheKey = $message;
            $module   = 'supplier';
            if (!empty($cacheKey)) {
                $data = (new DataCache())->getRecordByHash(
                    $module . '-' . QueueConstants::QUEUE_CHACHE_NAME, $cacheKey);
                if (empty($data)) {
                    return;
                }
                $consumerService = new SupplierConsumerService();
                $consumerService->process($data['values'], $cacheKey, $messageId);
            }
        } catch (\Exception $ex) {
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 接收消息并调用业务处理
     * @param array $messageBody
     */
    protected function successed(\iQueue\MessageBody $messageBody)
    {
        // 获取消息内容
        $message = $messageBody->getMessage();
        $messageId = $messageBody->getMessageId();
        $jsonMsg = (is_array($message) || is_object($message)) ?  json_encode($message) : $message;
        echo $messageId .PHP_EOL;
        echo $jsonMsg;
        // 执行脚本
        $cmd = sprintf("php %s %s '%s' %s", $this->getYii(), $this->queueAction, $jsonMsg, $messageId);
        $this->debug && print "正在执行:{$cmd}\n";
        $output = system($cmd);
        $this->debug && print "{$output}\n";
    }

    public function actionTest($supplierId)
    {
        $supplierId = '0478ecdf-558c-11e7-b8d2-7cd30abc935a';
        $supplierId = explode(',', $supplierId);
        $supplierConsumerService = new SupplierConsumerService();
        $supplierConsumerService->single($supplierId);
    }

    /**
     * 定义执行脚本时yii的路径
     * @return string
     */
    protected function getYii()
    {
        return dirname(dirname(dirname(__DIR__))) . '/yii';
    }
}